package cn.qio.channel;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

import cn.qio.channel.handler.ChannelHandler;
import cn.qio.channel.session.ChannelSession;
import cn.qio.event.Event;
import cn.qio.event.EventType;

/**
 * 选择器线程
 * 
 * @author chenhua
 *
 */
public class ThreadSelector extends Thread {

	// private final String selectorId;

	private final Selector selector;

	private volatile boolean toStop;

	private final ConcurrentLinkedQueue<Event> eventQueue;

	private final ConcurrentMap<String, ChannelSession> sessionMap;

	private final ConcurrentMap<SocketChannel, String> idChannelMap;

	public void setToStop(boolean toStop) {
		this.toStop = toStop;
	}

	public int getChannelSessionSize() {
		return sessionMap.size();
	}

	public Selector getSelector() {
		return selector;
	}

	public synchronized void addSessionMap(ChannelSession channelSession) {
		sessionMap.putIfAbsent(channelSession.getSessionId(), channelSession);
		idChannelMap.putIfAbsent(channelSession.getSocketChannel(), channelSession.getSessionId());
	}

	public ThreadSelector(Selector selector) {
		this.selector = selector;
		this.sessionMap = new ConcurrentHashMap<>();
		this.idChannelMap = new ConcurrentHashMap<>();
		this.eventQueue = new ConcurrentLinkedQueue<>();
	}

	@Override
	public void run() {
		// int seconds = 1000 * 1000 * 1000; //纳秒->秒
		// long spendOnEventHandle = 0;
		// long spendOnIoHandle = 0;
		while (selector.isOpen() && !toStop) {
			if (eventQueue.size() > 0) {
				doEventHandle(eventQueue.poll());
			} else {
				int readyChannelCnt;
				try {
					readyChannelCnt = selector.select(1000);
					if (readyChannelCnt == 0) {
						continue;
					}
					Set<SelectionKey> selectKeys = selector.selectedKeys();
					Iterator<SelectionKey> it = selectKeys.iterator();
					while (it.hasNext()) {
						SelectionKey skey = it.next();
						createThenFireEvent(skey);
						it.remove();
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	private void doEventHandle(Event event) {
		if (event == null) {
			return;
		}
		EventType eventType = event.getEventType();
		ArrayList<ChannelHandler> handlers = event.getChannelSession().getEventHandlers();
		if (eventType == EventType.CHANNEL_READ_READY) {
			ByteBuffer byteBuffer;
			Event exceptionEvent = null;
			try {
				byteBuffer = doRead(event);
				event.setData(byteBuffer);
			} catch (IOException e) {
				exceptionEvent = new Event(EventType.CHANNEL_EXCEPTION, event.getChannelSession(), null);
			}
			if (exceptionEvent != null) {
				event = exceptionEvent;
			}
			for (ChannelHandler handler : handlers) {
				handler.messageReceived(event);
			}
		} else if (eventType == EventType.CHANNEL_WRITE_READY) {
			for (ChannelHandler handler : handlers) {
				handler.messageReceived(event);
			}
		} else {
			throw new RuntimeException("not supported event type " + eventType);
		}

	}

	/**
	 * 读取当前缓冲区里面的字节，当前有多少读多少，或者读取缓冲区满了
	 * 
	 * @return
	 * @throws IOException
	 */
	private ByteBuffer doRead(Event event) throws IOException {
		ByteBuffer bb = ByteBuffer.allocate(2 * 1024); // 缓冲区一次最大读取2k
		boolean isStreamClosed = false;
		SocketChannel sc = event.getChannelSession().getSocketChannel();
		for (;;) {
			int readCnt = sc.read(bb);
			if (readCnt < 0) { // 返回0表示 对方关闭了链接
				isStreamClosed = true;
				break;
			} else if (readCnt == 0) {
				break;
			}
		}
		if (isStreamClosed) {
			doChannelClose(event);
		}
		bb.flip();
		return bb;
	}

	private void doChannelClose(Event event) {
		// session
		SelectionKey skey = event.getSelectKey();
		if (skey != null) {
			skey.cancel();
		}
		SocketChannel sc = event.getChannelSession().getSocketChannel();
		if (sc != null) {
			try {
				sc.close();
			} catch (IOException e) {
			}
		}
	}

	public void createThenFireEvent(SelectionKey skey) {
		EventType eventType = null;
		if (skey.isReadable()) {
			eventType = EventType.CHANNEL_READ_READY;
		} else if (skey.isWritable()) {
			eventType = EventType.CHANNEL_WRITE_READY;
		} else {
			throw new RuntimeException("not supported selection key " + skey.interestOps());
		}
		String channelId = idChannelMap.get(skey.channel());
		ChannelSession session = sessionMap.get(channelId);
		Event event = new Event(eventType, session, null);
		eventQueue.add(event);
	}

}
